package org.ponly.uno4j.core.impl;

import org.ponly.uno4j.core.UnoBridge;
import org.ponly.uno4j.core.UnoClient;
import org.ponly.uno4j.core.UnoTask;

import java.lang.ref.WeakReference;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.concurrent.*;

/**
 * 负载分发 UnoClient 实现
 *
 * @author vacoor
 */
public class LbUnoClientImpl implements UnoClient {
    /**
     * 存活的服务器
     */
    private final Map<String, ServerWrapper> aliveServers = new LinkedHashMap<String, ServerWrapper>();
    /**
     * 死亡的服务器
     */
    protected final Map<String, ServerWrapper> zombieServers = new ConcurrentHashMap<String, ServerWrapper>();

    /**
     * 存活的服务器队列
     */
    private LinkedBlockingQueue<ServerWrapper> aliveServerQueue = new LinkedBlockingQueue<ServerWrapper>();

    /**
     * 存活检查执行器
     */
    private ScheduledExecutorService aliveCheckExecutor;
    private int interval = 60 * 1000;

    public LbUnoClientImpl(String... unoUrls) {
        for (String unoUrl : unoUrls) {
            ServerWrapper wrapper = new ServerWrapper(makeServer(unoUrl));
            aliveServers.put(wrapper.getKey(), wrapper);
        }
        updateAliveQueue();
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public Future<?> submit(final UnoTask unoTask) {
        int maxTries = 10;

        for (int attempts = 0; attempts < maxTries; attempts++) {
            final ServerWrapper wrapper = acquire();
            wrapper.lastUsed = System.currentTimeMillis();

            try {
                return wrapper.unoServer.submit(new UnoTask() {
                    @Override
                    public void process(UnoBridge bridge) throws Exception {
                        try {
                            unoTask.process(bridge);
                        } finally {
                            addToAlive(wrapper);
                        }
                    }
                });
            } catch (Exception ex) {
                // move to Alive To dead(server)
                moveAliveToDead(wrapper);
            }
        }
        return null;
    }

    private ServerWrapper acquire() {
        ServerWrapper wrapper = null;
        try {
            wrapper = aliveServerQueue.poll(30, TimeUnit.SECONDS);
            // ServerWrapper pre = aliveServers.remove(wrapper.getKey());
        } catch (InterruptedException e) {
            throw new IllegalStateException(e);
        }
        return wrapper;
    }

    private void checkAZombieServer(ServerWrapper zombieServer) {
        long currTime = System.currentTimeMillis();
        try {
            zombieServer.lastChecked = currTime;
            // TODO PING
            if (true) {
                ServerWrapper wrapper = zombieServers.remove(zombieServer.getKey());
                if (null != wrapper) {
                    wrapper.failedPings = 0;
                    addToAlive(wrapper);
                }
            }
        } catch (Exception e) {
            //Expected. The server is still down.
            zombieServer.failedPings++;
        }
    }

    private void moveAliveToDead(ServerWrapper wrapper) {
        wrapper = removeFromAlive(wrapper.getKey());
        if (null == wrapper) {
            return;  // another thread already detected the failure and removed it
        }

        zombieServers.put(wrapper.getKey(), wrapper);
        startAliveCheckExecutor();
    }

    private ServerWrapper removeFromAlive(String key) {
        synchronized (aliveServers) {
            ServerWrapper wrapper = aliveServers.remove(key);
            if (wrapper != null) {
                updateAliveQueue();
            }
            return wrapper;
        }
    }

    private void updateAliveQueue() {
        synchronized (aliveServers) {
//            LinkedBlockingQueue<ServerWrapper> queue = new LinkedBlockingQueue<>();
            aliveServerQueue.clear();
            aliveServerQueue.addAll(aliveServers.values());
        }
    }

    private void addToAlive(ServerWrapper wrapper) {
        synchronized (aliveServers) {
            ServerWrapper prev = aliveServers.put(wrapper.getKey(), wrapper);
            // TODO: warn if there was a previous entry?
            updateAliveQueue();
        }
    }

    protected UnoClientImpl makeServer(String unoUrl) {
        return new UnoClientImpl(unoUrl);
    }

    public void shutdown() {
        ServerWrapper poll;
        while (null != (poll = aliveServerQueue.poll())) {
            poll.unoServer.shutdown();
        }
        if (null != aliveCheckExecutor) {
            aliveCheckExecutor.shutdown();
        }
    }

    protected static class ServerWrapper {
        final UnoClientImpl unoServer;

        long lastUsed;     // last time used for a real request
        long lastChecked;  // last time checked for liveness

        // "standard" servers are used by default.  They normally live in the alive list
        // and move to the zombie list when unavailable.  When they become available again,
        // they move back to the alive list.
        boolean standard = true;

        int failedPings = 0;

        public ServerWrapper(UnoClientImpl unoServer) {
            this.unoServer = unoServer;
        }

        @Override
        public String toString() {
//            return server.unoBridge.getBridgeName();
            return null;
        }

        public String getKey() {
            return unoServer.getUnoUrl();
        }

        @Override
        public int hashCode() {
            return this.getKey().hashCode();
        }

        @Override
        public boolean equals(Object obj) {
            if (this == obj) return true;
            if (!(obj instanceof ServerWrapper)) return false;
            return this.getKey().equals(((ServerWrapper) obj).getKey());
        }
    }

    private void startAliveCheckExecutor() {
        // double-checked locking, but it's OK because we don't *do* anything with aliveCheckExecutor
        // if it's not null.
        if (aliveCheckExecutor == null) {
            synchronized (this) {
                if (aliveCheckExecutor == null) {
//                    aliveCheckExecutor = Executors.newSingleThreadScheduledExecutor(new SolrjNamedThreadFactory("aliveCheckExecutor"));
                    aliveCheckExecutor = Executors.newSingleThreadScheduledExecutor();
                    aliveCheckExecutor.scheduleAtFixedRate(
                            getAliveCheckRunner(new WeakReference<LbUnoClientImpl>(this)),
                            this.interval, this.interval, TimeUnit.MILLISECONDS
                    );
                }
            }
        }
    }

    private static Runnable getAliveCheckRunner(final WeakReference<LbUnoClientImpl> lbRef) {
        return new Runnable() {
            @Override
            public void run() {
                LbUnoClientImpl lb = lbRef.get();
                if (lb != null && lb.zombieServers != null) {
                    for (ServerWrapper zombieServer : lb.zombieServers.values()) {
                        lb.checkAZombieServer(zombieServer);
                    }
                }
            }
        };
    }
}
